package com.cmnit.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cmnit.utils.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.DecimalFormat;
import java.util.List;

/**
 * 获取Hdfs资源
 */
public class HdfsService {

    /**
     * 获取HDFS页面资源
     */
    public static void getHdfsResources() {
        // 获取当前时间
        String time = DateUtils.getTimeString();
        // http地址
        String get = HttpUtils.doGet(ConfigurationManager.getProperty("http.hdfs.web"));
        JSONObject jsonObject = JSON.parseObject(get);
        // 获取hdfs全部资源情况
        String bean = jsonObject.getString("beans");
        JSONArray objects = JSON.parseArray(bean);
        // 目标表名（MySql）
        String tableName = "datalake_workspace_hdfstatics";
        // 目标表需要写入的表字段
        String updateField = "CapacityTotal,CapacityUsed,CapacityRemaining,FilesTotal,BlocksTotal,time";
        // 获取指定字段
        for (Object obj : objects) {
            JSONObject jsonOb = (JSONObject) obj;
            // HDFS空间总量
            String capacityTotal = jsonOb.getString("CapacityTotal");
            // HDFS空间已使用量
            String capacityUsed = jsonOb.getString("CapacityUsed");
            // HDFS空间剩余量
            String capacityRemaining = jsonOb.getString("CapacityRemaining");
            // HDFS文件总数
            String filesTotal = jsonOb.getString("FilesTotal");
            // Block总数
            String blocksTotal = jsonOb.getString("BlocksTotal");
            // 写入字段的值
            String valuesField = capacityTotal + "," + capacityUsed + "," + capacityRemaining + "," + filesTotal + "," + blocksTotal + "," + time;
            MySqlUtils.insert(tableName, updateField, valuesField);
        }
    }

    /**
     * 获取指定路径的hdfs大小
     */
    public static void getHdfsSize(String time) {
        // 获取账期时间
        String acctTime = DateUtils.getAcctTime(time);
        // 获取拼接分区字段
        String hdfsPartition = DateUtils.getHdfsPartition(time);

        // 目标表名（MySql）
        String tableName = "datalake_workspace_hdfshiveincrement";
        // 目标表需要写入的表字段
        String updateField = "data_source,data_name,size,time";

        // 读取数据库中的配置文件
        String sql = "select source_name, address, status from DIM.DIM_DATALAKE_DISPLAY where filed_sign = 'hdfs'";
        String resultField = "source_name,address,status";
        FileSystem fs = null;
        List<String> tableList = HiveUtils.query(sql, resultField);
        try {
            fs = FileSystem.get(new URI("hdfs://mycluster"), new Configuration());

            // 遍历所有的表或数据库地址
            for (String info : tableList) {
                String[] tableValue = info.split(",");
                // 名称
                String name = tableValue[0];
                // 地址
                String address = tableValue[1];
                // 类别（数据库或数据表）
                String dataSource = tableValue[2];
                // 计算结果保留2位小数
                DecimalFormat decimalFormat = new DecimalFormat("0.00");

                // 数据表地址需要拼接分区字段
                if ("table".equals(dataSource)) {
                    address += hdfsPartition;
                }

                try {
                    // 获取文件大小，并将B转为G
                    String size = decimalFormat.format((double) fs.getContentSummary(new Path(address)).getLength() / (double) 1073741824);
                    // Mysql结果表写入字段的值
                    String valuesField = dataSource + "," + name + "," + size + "," + acctTime;
                    MySqlUtils.insert(tableName, updateField, valuesField);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        } catch (IOException | URISyntaxException e) {
            e.printStackTrace();
        } finally {
            try {
                if (fs != null) {
                    fs.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
